package cn.doitedu.rtdw.yanggekun;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

@Slf4j
public class FreightHouseOrderType {

    public static void main(String[] args) throws Exception {

        System.setProperty("java.io.tmpdir", "d:/java_tmp/");

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);

        // 状态后端
        //env.setStateBackend(new EmbeddedRocksDBStateBackend());

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("freight-orders")
                .setGroupId("g" + System.currentTimeMillis())
                .setStartingOffsets(OffsetsInitializer.latest())
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setBootstrapServers("doitedu:9092")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> ds = env.fromSource(source, WatermarkStrategy.noWatermarks(), "dwd-kfk");

        // {"mysql_op_type":"c","acq_id":"243100000166954202","id":"1657232507873611776","warehouse_code":"W0P8","job_type":"1","stock_batch_id":"838105450456526848","store_location_code":"BZ0201081501","product_code":"1174251","product_name":"果子王爆浆球面包2.5kg","product_bar_code":"2100000867417","unit":"箱","production_date":"2023-01-31","expiry_date":"2023-05-01","product_days":"90.0","qty":"1.0","stock_use_attr":"1","create_by":"sys-job","create_time":"2023-05-13 11:53:14","update_by":"sys-job","update_time":"2023-05-13 11:53:14","purchase_group_code":"A02","purchase_group_name":"休闲食品课"}
        SingleOutputStreamOperator<Order> beans = ds.map(json -> JSON.parseObject(json, Order.class));

        SingleOutputStreamOperator<String> result = beans
                .keyBy(Order::getReceive_location_code)
                .process(new KeyedProcessFunction<String, Order, Accumulator>() {

                    MapState<String, Order> lastUpdateStatusState;
                    ValueState<Accumulator> accumulatorValueState;
                    ValueState<Long> timerState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastUpdateStatusState = getRuntimeContext().getMapState(new MapStateDescriptor<>("lastUpdateStatus", String.class, Order.class));

                        accumulatorValueState = getRuntimeContext().getState(new ValueStateDescriptor<Accumulator>("accumulator", Accumulator.class));
                        timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer", Long.class));
                    }

                    @Override
                    public void processElement(Order order, KeyedProcessFunction<String, Order, Accumulator>.Context context, Collector<Accumulator> collector) throws Exception {
                        // 初次注册定时器： 以启动时间为准，按分钟取整后，+1分钟作为第一次将输出结果的时间
                        if (timerState.value() == null) {
                            long outTime = (context.timerService().currentProcessingTime() / 10000) * 10000 + 10000;
                            System.out.println("注册的输出时间为:" + DateFormatUtils.format(outTime,"yyyy-MM-dd HH:mm:ss"));
                            context.timerService().registerProcessingTimeTimer(outTime);
                            timerState.update(outTime);
                        }

                        // 获取当前订单id对应的上一次订单数据
                        Order lastOrderStatus = lastUpdateStatusState.get(order.getId());
                        if (lastOrderStatus == null) {
                            lastOrderStatus = order;
                            lastUpdateStatusState.put(order.getId(), lastOrderStatus);
                        }

                        // 获取累加器并初始化处理
                        Accumulator accumulator = accumulatorValueState.value();
                        if (accumulator == null) {
                            accumulator = new Accumulator();
                            accumulator.setReceive_location_code(context.getCurrentKey());
                            accumulatorValueState.update(accumulator);
                        }

                        //   1. 未关单处理
                        // 如果 此前状态不满足未关单， 而本次满足: 未关单条件
                        // 则需要增加累计值，并将  order 未关单状态置为 true
                        if ((lastOrderStatus.getWeiGuan() == null || !lastOrderStatus.getWeiGuan()) && JudgeUtils.judgeWeiGuan(order)) {
                            accumulator.setWeiGuan(accumulator.getWeiGuan() + 1);
                            lastOrderStatus.setWeiGuan(true);
                            log.error("1. 未关单+1");
                        }

                        // 如果 该订单此前状态满足未关单，而本次不满足未关单，
                        // 则需要减累计值，并将 order未关单状态置为 false
                        if (lastOrderStatus.getWeiGuan() != null && lastOrderStatus.getWeiGuan() && !JudgeUtils.judgeWeiGuan(order)) {
                            accumulator.setWeiGuan(accumulator.getWeiGuan() - 1);
                            lastOrderStatus.setWeiGuan(false);
                            log.error("1. 未关单-1");
                        }


                        //   2. 未清入库单
                        // 如果 此订单是第一次出现 或 此前状态不满足 未清单， 而本次满足: 未清单条件
                        // 则需要增加累计值，并将  order 未清单状态置为 true
                        if ((lastOrderStatus.getWeiQing() == null || !lastOrderStatus.getWeiQing()) && JudgeUtils.judgeWeiQing(order)) {
                            accumulator.setWeiQing(accumulator.getWeiQing() + 1);
                            lastOrderStatus.setWeiQing(true);
                            log.error("2. 未清单 + 1");
                        }

                        // 如果 该订单此前状态满足未清单，而本次不满足未清单，
                        // 则需要减累计值，并将 order未清单状态置为 false
                        if (lastOrderStatus.getWeiQing() != null && lastOrderStatus.getWeiQing() && !JudgeUtils.judgeWeiQing(order)) {
                            accumulator.setWeiQing(accumulator.getWeiQing() - 1);
                            lastOrderStatus.setWeiQing(false);
                            log.error("2. 未清单 - 1");
                        }


                        //   3. 调拨入库单
                        // 如果 此订单是第一次出现 或 此前状态不满足 调拨单， 而本次满足: 调拨单单条件
                        // 则需要增加累计值，并将  order 调拨单状态置为 true
                        if ((lastOrderStatus.getDiaoBo() == null || !lastOrderStatus.getDiaoBo()) && JudgeUtils.judgeDiaoBo(order)) {
                            accumulator.setDiaoBo(accumulator.getDiaoBo() + 1);
                            lastOrderStatus.setDiaoBo(true);
                            log.error("3. 调拨单 + 1");
                        }

                        // 如果 该订单此前状态满足未关单，而本次不满足未关单，
                        // 则需要减累计值，并将 order未关单状态置为 false
                        if (lastOrderStatus.getDiaoBo() != null && lastOrderStatus.getDiaoBo() && !JudgeUtils.judgeDiaoBo(order)) {
                            accumulator.setDiaoBo(accumulator.getDiaoBo() - 1);
                            lastOrderStatus.setDiaoBo(false);
                            log.error("3. 调拨单 - 1");
                        }

                        //   4. 待上架任务
                        // 如果 此单是第一次出现 或 此前状态不满足， 而本次满足
                        // 则需要增加累计值，并将  order 待上架任务单 状态置为 true
                        if ((lastOrderStatus.getDaiShangJia() == null || !lastOrderStatus.getDaiShangJia()) && JudgeUtils.judgeDaishangJia(order)) {
                            accumulator.setDaiShangjia(accumulator.getDaiShangjia() + 1);
                            lastOrderStatus.setDaiShangJia(true);
                            log.error("4. 待上架 + 1");
                        }

                        // 如果 该订单此前状态满足未关单，而本次不满足未关单，
                        // 则需要减累计值，并将 待上架任务单 状态置为 false
                        if (lastOrderStatus.getDaiShangJia() != null && lastOrderStatus.getDaiShangJia() && !JudgeUtils.judgeDaishangJia(order)) {
                            accumulator.setDaiShangjia(accumulator.getDaiShangjia() - 1);
                            lastOrderStatus.setDaiShangJia(false);
                            log.error("4. 待上架 - 1");
                        }


                        //   5. 客户退单任务
                        // 如果 此单是第一次出现 或 此前状态不满足， 而本次满足
                        // 则需要增加累计值，并将  order 退单 状态置为 true
                        if ((lastOrderStatus.getTuiDan() == null || !lastOrderStatus.getTuiDan()) && JudgeUtils.judgeTuidan(order)) {
                            accumulator.setTuiDan(accumulator.getTuiDan() + 1);
                            lastOrderStatus.setTuiDan(true);
                            log.error("5. 退单 + 1");
                        }

                        // 如果 该订单此前状态满足未关单，而本次不满足未关单，
                        // 则需要减累计值，并将 待上架任务单 状态置为 false
                        if (lastOrderStatus.getTuiDan() != null && lastOrderStatus.getTuiDan() && !JudgeUtils.judgeTuidan(order)) {
                            accumulator.setTuiDan(accumulator.getTuiDan() - 1);
                            lastOrderStatus.setTuiDan(false);
                            log.error("5. 退单 - 1");
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, Order, Accumulator>.OnTimerContext ctx, Collector<Accumulator> out) throws Exception {
                        Accumulator res = accumulatorValueState.value();
                        res.setOutTime(timestamp);
                        out.collect(res);

                        ctx.timerService().registerProcessingTimeTimer(timestamp + 10000);
                        timerState.update(timestamp + 10000);
                    }

                })
                .map(JSON::toJSONString)  // 转成json输出方便查看
                ;

        result.print();

        env.execute();
    }
}
